<?php
/* vim: set expandtab tabstop=4 shiftwidth=4 encoding=utf-8: */
/* $Id: */
/*
Copyright (c) 2010, Justin Swanhart
All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
    * Redistributions of source code must retain the above copyright
      notice, this list of conditions and the following disclaimer.
    * Redistributions in binary form must reproduce the above copyright
      notice, this list of conditions and the following disclaimer in the
      documentation and/or other materials provided with the distribution.
    * Neither the name of the <organization> nor the
      names of its contributors may be used to endorse or promote products
      derived from this software without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY
DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/

/*This script requires PEAR Net_Gearman */
/*It also requires Console_Getopt, but this should be installed by default with pear */
require_once 'common.php';
#$params = get_commandline();

class ShardQuery {
	var $parser = false;
	var $messages = array();
	var $verbose;
    var $info = array();

	var $workers; # the count of tasks
	var $shard_key = false;

	var $col_metadata_cache=array();
	var $table_is_sharded_cache=array();

    function __get($propname) {
        if(isset($this->$propname)) return($this->$propname);
        if(isset($this->state) && isset($this->state->$propname)) return($this->state->$propname);
    }

    static function new_state() {
        $state = new StdClass;

        $state->insert_values = "";
        $state->error = false;
        $state->force_shard = array();
        $state->table_name = "";
        $state->tables = array();
        $state->DAL = null;
        $state->agg_key_cols="";
        $state->non_distrib = false;
        $state->total_time=0;
        $state->exec_time=0;
        $state->insert_sql = "";
        $state->used_colrefs = array();
        $state->coord_odku = array();
        $state->subqueries = array();
        $state->broadcast_query = false;
        $state->tmp_shard = null;
        $state->client = null;

        $state->store_resultset_count = 0;
        $state->function_count = 0;

        return $state;
    }

	function __construct($schema_name = null, $config_file = 'shard-query-config.php') {
		$this->parser = new PHPSQLParser();
		ShardQueryCustomFunctions::register_all($this->parser);
		$this->state = $this->set_schema($schema_name, $state = ShardQuery::new_state(), $config_file);
	}

	function set_schema($schema_name = null,&$state = null, $config_file = 'shard-query-config.php') {
        if(!isset($state)) $state = ShardQuery::new_state();

        $state->schema_name = $schema_name;

		require($config_file);
		if(!$schema_name && isset($config_database['default_virtual_schema'])) $schema_name = $config_database['default_virtual_schema'];

		/* reset defaults */
		$state->verbose=false;
		$state->inlist_opt="*";
		$state->between_opt="*";
		$state->gearman_servers = array();
		$state->inlist_merge_threshold=128;
		$state->inlist_merge_size=128;
		$state->current_schema = $schema_name;
		$state->extra_tables[] = array(); #any tables that get created along the way and need cleaning up at the end

		switch($config_database['mapper_type']) {
			case SQ_DIRECTORY:
				$mapper = new DirectoryShardKeyMapper($config_database);	
			break;

			case SQ_HASH:
				$mapper = new HashShardKeyMapper($config_database);	
			break;

			default:
				throw new Exception('Unsupported mapper: ' . $config_database['mapper']);
			break;
		}

		$params = $mapper->get_params($schema_name);
		$shards = $mapper->get_shards($schema_name);

		$state->shards = $shards;

        $state->mapper = &$mapper;

		if(in_array('verbose', array_keys($params))) {
			$state->verbose = true;
		}

		if(!empty($params['inlist_merge_threshold'])) {
			$state->inlist_merge_threshold=$params['inlist_merge_threshold'];
		}

		if(!empty($params['inlist_merge_size'])) {
			$state->inlist_merge_size=$params['inlist_merge_size'];
		}

		$state->engine='InnoDB';
		if(!empty($params['coord_engine'])) {
			$state->engine=$params['coord_engine'];
		}

		$this->set_shard_mapper($column = $mapper->get_column($schema_name), array($mapper,'map'));


        #FIXME: The loader is probably quite broken at the moment.  Fix the loader :(
		if(!empty($params['chunk'])) $this->chunk = $params['chunk']; else $this->chunk = false;
	
		/* One shard acts as the coordination shard.  A table is created on this shard
		   to coallesce the results from the other shards.  This shard is used when a query broadcast
		   is necessary, otherwise only one shard will be used.
		*/

		$coords = $mapper->get_shards($schema_name, true);
		if(!empty($coords)) { 
			$state->tmp_shard = $coords[array_rand($coords)];
		} else {
			$state->tmp_shard = $shards[array_rand($shards)];
		}
	
		if(!empty($params['inlist'])) {
			$state->inlist_opt = trim($params['inlist']);
			if($state->inlist_opt == "-") $state->inlist_opt = $column;
		}

		if(!empty($params['between'])) {
			$state->between_opt = trim($params['between']);
			if($state->inlist_opt == "-") $state->inlist_opt = $column;
		}

		$state->gearman_servers=$mapper->get_gearman_servers($schema_name);

        return $state;
	
	}

	function query($sql, $explain_only = false, $alt_state = null, $keep_result = false) {
        if($this->async) $keep_result = true;
        if(isset($alt_state)) $state = $alt_state; else $state = $this->state;
        
		$this->hashes = array();
		$state->custom_functions = array();
		$this->errors=array();
		$state->info = array();
		$state->orig_sql = $sql;

		$state->insert_sql = "";
		$state->insert_values = array();
		$state->got_rows = 0;
		$state->messages=array();
	
		$start = microtime(true);
		$state->DAL = SimpleDAL::factory($state->tmp_shard);

		if($state->DAL->my_error()) {
			throw new Exception("connect to the coordinator DB:" . $state->DAL->my_error() . "\n");
		}
		
		if(!$state->DAL->my_select_db($state->tmp_shard['db'])) {
			throw new Exception("use the coord db database:" . $state->DAL->my_error());
		} 
		$this->workers=0;

		if(!$state->client = new Net_Gearman_Client($state->gearman_servers)) throw new Exception('Could not instantiate gearman client\n');
		$state->subqueries = array();

		if(!$this->process_sql($sql, $state)) {
			return false;	
		}

		if($explain_only) return $state->explain;

		$state->total_time = $state->parse_time = microtime(true) - $start;
		$start = microtime(true);

		if($state->insert_sql) {	
			$result_count=0;
			foreach($state->insert_values as $shard_id => $_vals) {
				$insert_sql = $state->insert_sql . $_vals;
				$state->DAL = SimpleDAL::factory($state->shards[$shard_id]);
				$state->DAL->my_select_db($state->shards[$shard_id]['db']);
				$result = $state->DAL->my_query($insert_sql);
				if($result) $result_count += $state->DAL->my_affected_rows(); else throw new Exception($state->DAL->my_error());
			}

		
			$sql = "SELECT 'INSERT statement processed.  Inserted {$result_count} rows.' as result";
			$stmt = $state->DAL->my_query($sql);
			$state->exec_time = microtime(true) - $start;
			$state->total_time += $state->exec_time;
			return($stmt);	
		}

		if($state->broadcast_query !== false) {
			$result = true;
			$this->broadcast($state->broadcast_query, array($this, 'generic_callback'), null, 'no_resultset');
			$this->wait_for_workers();
			if($this->errors) $result = false;
	
			$state->exec_time = microtime(true) - $start;
			$state->total_time += $state->exec_time;
			return($result);	
		}


		$collect_cb = array($this,'generic_callback');
		$failure_cb = array($this,'handle_failure');

		#subqueries in the from clause get parallelized on each storage node
		if($this->errors || $this->errors) {
			return false;
		} 

        $job_id = null;

        if($this->async) $job_id = $state->mapper->register_job( 
                                                        $state->tmp_shard['id'],
                                                        0, 
                                                        0, 
                                                        "''", 
                                                        $state->coord_sql);

        $state->query_id = $job_id;

        if(!empty($state->shard_sql)) {

		    $set = new Net_Gearman_Set();
		    $this->workers=0;

		    $this->create_gearman_set($state->shard_sql, $state->table_name, $collect_cb, $failure_cb,$set, "store_resultset", $state);

		    $this->run_set($set, $state);
		    $this->wait_for_workers();

		    if($this->errors) {
		    	return false;
		    } 

		    $stmt = false;

		    unset($set);

		    $this->errors = array();

		    if($state->got_rows) {
		    	if(isset($state->custom_functions) && !empty($state->custom_functions)) {

		    		$set = new Net_Gearman_Set();

		    		foreach($state->custom_functions as $func_call_id => $function) {
		    			if(!empty(ShardQueryCustomFunctions::$list[$function['function']]) && ShardQueryCustomFunctions::$list[$function['function']]['sort_input'] === true) {

		    				$key = "(gb_hash";
		    				foreach($function['colref_map'] as $col) {

		    					$key .= ", $col";
		    				}
		    				$key .= ")";
	
		    				$sql = "ALTER TABLE `{$state->table_name}` ADD KEY $key";

		    				$state->DAL->my_query($sql);
		    			}
		    				
                        if($this->async) {
                            $task_type = Net_Gearman_task::JOB_BACKGROUND;
                        } else {
                            $task_type = Net_Gearman_task::JOB_NORMAL;
                        }
		    			foreach($this->hashes as $hash  => $discard) {
	  	    				$task = new Net_Gearman_Task(   'custom_function',
		    								array('func_call_id' => $func_call_id, 'function_info' => $function, 'tmp_shard' => $state->tmp_shard, 'table' => $state->table_name, 'gb_hash' => $hash ),
		    								uniqid() . mt_rand(1, 10000000),
                                            $task_type
		    				);
                            $state->function_count++;
                            $state->extra_tables[] = $func_call_id;

		    				$task->attachCallback($collect_cb);
		    				$task->attachCallback($failure_cb,Net_Gearman_Task::TASK_FAIL);

		    				$set->addTask($task);
		    			}
		    	
		    		}

		    		
		    		$this->run_set($set, $state);
		    		$this->wait_for_workers();

		    		if(!empty($this->errors)) {
		    			$this->cleanup_tables($state);
		    			return false;
		    		}

		    	}

                if(!$keep_result) {
		    	    $state->DAL = SimpleDAL::factory($state->tmp_shard);	
		        	$stmt = $this->aggregate_result($state);
                } else {
                    $stmt = true;
                }

		    }

        } else {
                if(!$keep_result) {
		    	    $state->DAL = SimpleDAL::factory($state->tmp_shard);	
		        	$stmt = $this->aggregate_result($state);
                } else {
                    $stmt = true;
                }
        }
        $shard_id = $state->tmp_shard['id'];
        
        $table_list = $state->extra_tables;
        $table_list[] = $state->table_name;

        /* return a job_id if this is an async job */
        if($this->async) { 
            $state->mapper->update_job( $job_id, $state->store_resultset_count, $state->function_count, "", $state->coord_sql );
            $stmt = $job_id;
        }
        

        $this->DAL = $state->DAL;

		$state->exec_time = microtime(true) - $start;
		$state->total_time += $state->exec_time;

		return $stmt;
		
	}

	function wait_for_workers() {
        if($this->async) return true;
		do {
			usleep(1000);
		} while ($this->workers > 0);

		return true;
	}



	#this function broadcast one or more queries of any type to all shards
	function broadcast($queries,$collect_cb=null,$failure_cb=null,$action="no_resultset") {

		if(!$collect_cb) $collect_cb=array($this,'generic_callback');
		if(!$failure_cb) $failure_cb=array($this,'handle_failure');
		if(!is_array($queries)) $queries = array( $queries );

		$set = new Net_Gearman_Set();

		$this->run_set($this->create_gearman_set($queries, false, $collect_cb, $failure_cb,$set, $action), $state);
		
	}


	protected function process_select_item($pos, 
						&$clause,
						&$shard_query="", 
						&$coord_query="", 
						&$push_select=array(), 
						&$group_aliases=array(), 
						&$group=array(),
						&$error=array(),
						$skip_alias = false,
						&$used_agg_func,
						&$coord_odku,
						$prev_clause = null,
                        &$state
						) {
	
        $return = array();	
        $no_pushdown_limit = true;
		$non_distrib = false;
        $alias = "";

		if(!empty($clause['base_expr']) && $clause['base_expr'] == "*") {
			$shard_query .= "*";
			$coord_query .= "*";
			return true;
		}

		if(empty($clause['alias']) || $clause['alias'] == "") {
			$item = "";
			if($clause['expr_type'] != 'expression') {
				$alias = "`" . $this->concat_all_subtrees($clause,$item) . "`";
			} else {
				$alias = "`{$clause['base_expr']}`";
			}
		} else {
			$alias = "`" . trim($clause['alias']['name'],"`") . "`";
		}

		if(strpos($alias,'.')) {
			$alias=trim($alias,'``');
			$alias = explode('.', $alias);
			$alias = $alias[1];
			$alias="`$alias`";
		}

		if(!empty($clause['base_expr'])) {
			$base_expr = $clause['base_expr']; 
		}

		switch($clause['expr_type']) {
			case 'expression':
				if(!empty($clause['alias'])) {
					$alias = $clause['alias']['name'];
				}
				foreach($clause['sub_tree'] as $sub_pos => $sub_clause) {
						if($sub_pos > 0 ) {
							$prev_clause = $clause['sub_tree'][$sub_pos-1];
							if ($clause['sub_tree'][$sub_pos-1]['expr_type'] == 'operator' || $clause['sub_tree'][$sub_pos-1]['expr_type'] == 'reserved') {
								if($coord_query) $coord_query .= " ";
								if($shard_query) $shard_query .= " ";
							}  else {
								if($coord_query) $coord_query .= " ";
								if($shard_query) $shard_query .= " ";
							}
						} else {
							$prev_clause = $clause;
						}

						$this->process_select_item($pos, 
							$sub_clause,
							$shard_query,
							$coord_query,
							$push_select,
							$group_aliases,
							$group,
							$error,
							true,
							$used_agg_func,
							$coord_odku,
                            $prev_clause
						);

					if($sub_pos == count($clause['sub_tree'])-1) {
						$coord_query .= " AS $alias";	
					}
					
				}	
			break;

			case 'custom_function':
				$no_pushdown_limit = true;
				$non_distrib = true;
               
			break;

	
			case 'aggregate_function':
				$used_agg_func=1;
				$item="";	
				$base_expr = $this->concat_all_subtrees($clause['sub_tree'],$item);

				$function = strtoupper($clause['base_expr']);
			
				switch($function) {
					case 'MIN':
					case 'MAX':
					case 'SUM':
						$expr_info=explode(" ", $base_expr);
						if(!empty($expr_info[0]) && strtolower($expr_info[0]) == 'distinct') {
								
								$this->messages[] = "Detected a {$function} [DISTINCT] expression!\n";
								$no_pushdown_limit = true;
								
								unset($expr_info[0]);
								$new_expr = join(" ", $expr_info);
								if(empty($state->used_colrefs[trim($new_expr)])) {
									if($shard_query) {
										if($prev_clause && !in_array($prev_clause['expr_type'], array('operator','function','aggregate_function','reserved'))) {
											$shard_query .= " ";
										} else {
											$shard_query .= ",";
										}
									}
									$shard_query .= "$new_expr AS $alias";
									$coord_odku[] = "$alias=VALUES($alias)";
									$group_aliases[] = $alias;
								}
								if($coord_query) {
									if($prev_clause && !in_array($prev_clause['expr_type'], array('operator','function','aggregate_function','reserved'))) {
										$coord_query .= " ";
									} else {
										$coord_query .= ",";
									}
								}
								$coord_query .= "{$function}(distinct $alias)" . (!$skip_alias ? " AS $alias" : "");	
						} else {
							switch($function) {
								case 'SUM':
									$coord_odku[] = "$alias=$alias +  VALUES($alias)";
									break;
								case 'MIN':
									$coord_odku[] = "$alias=IF($alias < VALUES($alias), VALUES($alias),$alias)";
									break;
								case 'MAX':
									$coord_odku[] = "$alias=IF($alias > VALUES($alias), VALUES($alias), $alias)";
									break;
							}

							if($shard_query) {
								if($prev_clause && !in_array($prev_clause['expr_type'], array('operator','function','aggregate_function','reserved'))) {
									$shard_query .= " ";
								} else {
									$shard_query .= ",";
								}
							}

							if($coord_query) {
								if($prev_clause && !in_array($prev_clause['expr_type'], array('operator','function','aggregate_function','reserved'))) {
									$coord_query .= " ";
								} else {
									if($pos > 1) $coord_query .= ",";
								}
							} 
							$shard_query .= "{$function}({$base_expr}) AS $alias";
							$coord_query .= "{$function}({$alias})" . (!$skip_alias ? " AS $alias" : "");
						}
						
					break;
					
					case 'AVG':								
						$expr_info=explode(" ", $base_expr);
						if(!empty($expr_info[0]) && strtolower($expr_info[0]) == 'distinct') {
								$this->messages[] = "Detected a {$function} [DISTINCT] expression!\n";
								
								unset($expr_info[0]);
								$new_expr = join(" ", $expr_info);
								if(empty($state->used_colrefs[trim($new_expr)])) {
									if($shard_query) {
										if($prev_clause && !in_array($prev_clause['expr_type'], array('operator','function','aggregate_function','reserved'))) {
											$shard_query .= " ";
										} else {
											$shard_query .= ",";
										}
									}
									$shard_query .= "$new_expr AS $alias";
									$coord_odku[] = "$alias=VALUES($alias)";
									$group_aliases[] = $alias;
								}
								if($coord_query) $coord_query .= ",";
								$coord_query .= "{$function}(distinct $alias)" . (!$skip_alias ? " AS $alias" : "");
								#since we pushed a GB expression we need to update the ODKU clause
						} else {
								$alias_cnt = trim($alias,'`');
								$alias_cnt = "`{$alias_cnt}_cnt`";
								if($shard_query) {
									if($prev_clause && !in_array($prev_clause['expr_type'], array('operator','function','aggregate_function','reserved'))) {
										$shard_query .= " ";
									} else {
										$shard_query .= ",";
									}
								}
								$shard_query .= "SUM({$base_expr}) AS $alias";
								if($coord_query) {
									if($prev_clause && !in_array($prev_clause['expr_type'], array('operator','function','aggregate_function','reserved'))) {
										$coord_query .= " ";
									} else {
										$coord_query .= ",";
									}
								}
								$coord_query .= "SUM({$alias})/SUM({$alias_cnt})" . (!$skip_alias ? " AS $alias" : "");

								#need to push a COUNT expression into the SELECT clause
								$push_select[] = "COUNT({$base_expr}) as {$alias_cnt}";
								$coord_odku[] = "{$alias_cnt}={$alias_cnt} +  VALUES({$alias_cnt})";

						}
						

					break;

					case 'STDDEV':
					case 'STD':
					case 'STDDEV_POP':
					case 'STDDEV_SAMP':
					case 'VARIANCE':
					case 'VAR_POP':
					case 'VAR_SAMP':
					case 'GROUP_CONCAT':
						$no_pushdown_limit = true;
						$non_distrib = true; 
						break 2;

					break;

					case 'COUNT':
						$base_expr=trim($base_expr,' ()');

						$expr_info=explode(" ", $base_expr);
						if(!empty($expr_info[0]) && strtolower($expr_info[0]) == 'distinct') {
							$this->messages[] = "Detected a COUNT [DISTINCT] expression!\n";
							$no_pushdown_limit = true;

							unset($expr_info[0]);
							$new_expr = join(" ", $expr_info);

							if(empty($state->used_colrefs[trim($new_expr)])) {
								if($shard_query) {
									if($prev_clause && !in_array($prev_clause['expr_type'], array('operator','function','aggregate_function','reserved'))) {
										$shard_query .= " ";
									} else {
										$shard_query .= ",";
									}
								}
								$shard_query .= "$new_expr AS $alias";
								$coord_odku[] = "$alias=VALUES($alias)";
								$group_aliases[] = $alias;
							}

							$state->used_colrefs[trim($new_expr)] = 1;
							if($coord_query) {
								if($prev_clause && !in_array($prev_clause['expr_type'], array('operator','function','aggregate_function','reserved'))) {
									$coord_query .= " ";
								} else {
									$coord_query .= ",";
								}
							}
							$coord_query .= "COUNT(distinct $alias)" . (!$skip_alias ? " AS $alias" : "");
							
						} else {

							if($shard_query) {
								if($prev_clause && in_array($prev_clause['expr_type'], array('operator','function','aggregate_function','reserved'))) {
									$shard_query .= " ";
								} else {
									$shard_query .= ",";
								}
							}

							if($coord_query) {
								if($prev_clause && in_array($prev_clause['expr_type'], array('operator','function','aggregate_function','reserved'))) {
									$coord_query .= " ";
								} else {
									if($pos > 1) $coord_query .= ",";
								}
							} 


							$shard_query .= "COUNT({$base_expr}) AS $alias";
							$coord_query .= "SUM($alias)" . (!$skip_alias ? " AS $alias" : "");
							$coord_odku[] = "$alias=$alias +  VALUES($alias)";
						}
	
					break;
		
					default:
						$error[] = array('error_clause' => $clause['base_expr'],
								'error_reason' => 'Unsupported aggregate function');
								
			
					break;
				}
		
			break;
	
			case 'operator':
			case 'reserved':
				$coord_query .= $base_expr;
				if($prev_clause && ($prev_clause['expr_type'] == 'expression' 
				|| $prev_clause['expr_type'] == 'function' 
				|| $prev_clause['expr_type'] == 'aggregate_function'
				|| $prev_clause['expr_type'] == 'colref' ) ) {
					$shard_query .= ",";
				} else {
				#	$shard_query .= $base_expr;
				} 
			
			break;


			case 'const':
				$coord_query .= $base_expr;
				#$shard_query .= $base_expr; #. " AS $alias";
			break;

			case 'function':
				$no_pushdown_limit = true;
				if($coord_query) $coord_query .= ",";
				$coord_query .= $base_expr . "(";
				$first = 0;
				foreach($clause['sub_tree'] as $sub_pos => $sub_clause) {
					if($sub_clause['expr_type'] == 'colref' || $sub_clause['expr_type'] == 'aggregate_function' || $sub_clause['expr_type'] == 'function') {
				#		$prev_clause = $clause;

						$this->process_select_item($pos+1, 
						$sub_clause,
						$shard_query,
						$coord_query,
						$push_select,
						$group_aliases,
						$group,
						$error,
						true,
						$used_agg_func,
						$coord_odku,
						$clause
						); //no aliases
					} else {
						if($sub_pos > 0 ) {
							if ($clause['sub_tree'][$sub_pos-1]['expr_type'] != 'operator' && $clause['sub_tree'][$sub_pos-1]['expr_type'] != 'reserved') {
								if($sub_clause['expr_type'] != 'operator') {
									if($coord_query) $coord_query .= ",";
								} else {
									if($coord_query) $coord_query .= " ";

								}
							}  else {
								if($coord_query) $coord_query .= " ";
							}
						}
						$coord_query .= $sub_clause['base_expr'];
					}
				}
				$coord_query .= ")";
				if($sub_pos == count($clause['sub_tree'])-1) {
					$coord_query .= " AS $alias";	
				}

				
			break;

			case 'colref':
			case 'subquery':
				if($prev_clause && !in_array($prev_clause['expr_type'], array('operator','function','aggregate_function','reserved'))) {
					if($coord_query) $coord_query .= ',';
				}
				$coord_query .= $alias;
				$coord_odku[] = "$alias=VALUES($alias)";
				if(empty($state->used_colrefs[$base_expr])) {
					$group[] = $pos+1;
					$group_aliases[] = $alias;
					if($prev_clause && !in_array($prev_clause['expr_type'], array('operator','function','aggregate_function','reserved'))) {
						if($shard_query) $shard_query .= ',';
					}
					$shard_query .= $base_expr . ' AS ' . $alias;
					$state->used_colrefs[$base_expr] = 1;
				} else {
					return false;
				}
		
			break;
	
			default:
	
				$this->errors[] = array('Unsupported expression:' => $clause['base_expr']);
				return false;
			break;
		}	

        $state->no_pushdown_limit = $no_pushdown_limit;
		$state->non_distrib = $non_distrib;

		return true;

	}

	protected function process_select($select, $straight_join=false,$distinct=false, &$state) {

		$error = array();	
		$shard_query=""; 		#Query to send to each shard
		$coord_query=""; 		#Query to send to the coordination node
	
		$group = array();		#list of positions which contain non-aggregate functions
		$group_aliases=array(); 	#list of group aliases which will be indexed on the aggregation temp table
						  
		$coord_odku=array();

		
		$used_agg_func=0;

		$push_select = array(); #list of expressions to push onto the end of the SELECT clause sent to the shards
		$state->non_distrib = false;

		$state->used_colrefs=array();
		$state->push_pos =0;
		
		$state->no_pushdown_limit = false; #will be set if a rewrite forces us to abandon pushdown limit strategy
		$prev_clause=false;
		foreach($select as $pos => $clause) {
			#this will recurse and fill up the proper structures
			if($coord_query && substr(trim($coord_query), -1) != ",") $coord_query .= ",";
/*
	protected function process_select_item($pos, 
						&$clause,
						&$shard_query="", 
						&$coord_query="", 
						&$push_select=array(), 
						&$group_aliases=array(), 
						&$group=array(),
						&$error=array(),
						$skip_alias = false,
						&$used_agg_func,
						&$coord_odku,
						$prev_clause = null,
                        &$state
*/
			$this->process_select_item(0, 
                                        $clause, 
                                        $shard_query, 
                                        $coord_query,
                                        $push_select, 
                                        $group_aliases, 
                                        $group, 
                                        $error, 
                                        false, 
                                        $used_agg_func,
                                        $coord_odku, 
                                        null,
                                        $state);

			$prev_clause = false;
		}

		$sql = "SELECT ";
		if($straight_join) $sql .= "STRAIGHT_JOIN ";
		if($distinct) $sql .= " DISTINCT ";

		$shard_query = $sql . $shard_query;
		$coord_query = $sql . $coord_query;

		foreach($push_select as $clause) {
			$shard_query .= "," . $clause;
		}
	
		$group_aliases = array_unique($group_aliases);
		if($used_agg_func) {
			$state->used_agg_func = true;
		} else {
			$state->used_agg_func = false;
			$group = array(); $group_aliases = array();
		}

		#we can't send pushed group by to the coord shard, so send the expression based 
		return array(
			'error'=>$error,
			'shard_sql'=>$shard_query, 
			'coord_odku' => $coord_odku, 
			'coord_sql'=> $coord_query,
			'shard_group' => join(',',$group_aliases), 
			'coord_group'=>join(',',$group), 
			'group_aliases' => join(',',$group_aliases)
		);
	}

	protected function process_undistributable_select_item(&$pos,&$real_pos,&$clauses, &$shard_query, &$coord_query,&$custom_functions, $func_call_id = null, &$state = null) {
		static $last_pos = null;
		static $is_group_item = false;
		static $is_custom_function = false;

		if($last_pos !== $pos) {
			$last_pos = $pos;
			$is_group_item = false;
		}

		if(empty($clauses[0])) {
			$clauses = array($clauses);
		}

		foreach($clauses as $clause) {

			switch($clause['expr_type']) {

				case 'colref':
					if($shard_query) $shard_query .= ",";
					$new_alias = "`p{$pos}_{$real_pos}`";
					++$real_pos;
					$shard_query .= $clause['base_expr'] . " AS $new_alias";
					if(!$is_custom_function) { 
						$coord_query .= $new_alias;
						if(!$func_call_id) $is_group_item = true;
					} else {
						$custom_functions[$func_call_id]['colref_map'][$clause['base_expr']] = $new_alias;
					}
				break;

				case 'custom_function':
					$is_custom_function = true;
					$func_call_id = md5(uniqid('',true));
					$custom_functions[ $func_call_id ] = array( 'function' => $clause['base_expr'], 'arg' => $clause['sub_tree'] );
					if($coord_query) $coord_query .= ",";
					$coord_query .= "MAX(( select retval from `$func_call_id` cfr where cfr.gb_hash = #gb_hash# limit 1 )) ";
					$this->process_undistributable_select_item($pos, $real_pos, $clause['sub_tree'], $shard_query, $coord_query,$custom_functions, $func_call_id, $state);
					$is_group_item = false;
					$func_call_id = null;
					$is_custom_function =false;
			
				break;

				case 'aggregate_function':
					#if($clause['expr_type'] == 'aggregate_function') $is_group_item = false;
					$is_group_item = false;
					if($clause['base_expr'] == 'count') {
						if($clause['sub_tree'][0]['base_expr'] == '*') {
							$clause['sub_tree'][0]['base_expr'] = "1";
						}
					} 
				case 'function':
					$func_call_id = md5(uniqid());
				case 'expression':

					if($coord_query) $coord_query .= ",";

					$coord_query .= "{$clause['base_expr']}(";
					$this->process_undistributable_select_item($pos, $real_pos, $clause['sub_tree'], $shard_query, $coord_query,$custom_functions, $func_call_id, $state);
					$coord_query .= ")";

					$func_call_id = false;
		
				break;

				case 'reserved':
				case 'operator':
				default:
					if(!isset($func_call_id)){ 
						$coord_query .= $clause['base_expr'] . " ";
					} 
						
				break;

			}
		}

		return $is_group_item;

	}

	protected function concat_all_subtrees($clauses, &$item) {

		if(empty($clauses[0])) {
			$clauses = array($clauses);
		}

		foreach($clauses as $clause) {
			
			switch($clause['expr_type']) {

				case 'function':
				case 'aggregate_function':
				case 'expression':
					$item .= "{$clause['base_expr']}(";
					$this->concat_all_subtrees($clause['sub_tree'], $item);
					$item .= ")";
				break;

				case 'reserved':
				case 'operator':
					$item .= $clause['base_expr'] . " ";
				break;

				default:
					$item .= $clause['base_expr'];
			}

		}

		return $item;

	}

	protected function process_having(&$state) {
		if(empty($state->parsed['HAVING'])) return false;

		$clauses = $state->parsed['HAVING'];
		unset($state->parsed['HAVING']);

		$having = "";
		$select = "";	

		foreach($clauses as $clause) {	
			$pos = 0;
			$shard = "";
			$coord = "";
			$push= array();
			$group_aliases=array();
			$group=array();
			$error=array();
			$skip_alias = true;
			$used_agg_func=false;
			$coord_odku=array();
			switch($clause['expr_type']) {
				case 'function':
				case 'aggregate_function':
				case 'expression':	
				if($select) $select .= ",";


				$this->process_select_item(
				$pos+1, 
				$clause,
				$select,
				$having,
				$push,
				$group_aliases,
				$group,
				$error,
				$skip_alias,
				$used_agg_func,
				$coord_odku,
                $state);

				break;

				default:
					$having .= $clause['base_expr'];
			}
		}

		if($having) $having= ' HAVING ' . $having . ' ';

		return array('having' => $having, 'select' => $select);
	}


	protected function process_undistributable_select($select, $straight_join=false,$distinct=false, &$state = null) {

		$shard_query=""; 		#Query to send to each shard
		$coord_query=""; 		#Query to send to the coordination node
	
		$state->no_pushdown_limit = true;
		$state->used_colrefs=array();
		$custom_functions = array();
		$group_items = "";
		foreach($select as $pos => $clause) {
			if(empty($clause['alias'])) {
				$alias = "";
				$alias = '`' . $this->concat_all_subtrees($clause, $alias) . '`';
			} else {
				$alias = "`" . $clause['alias']['name'] . "`";
			}
			#this will recurse and fill up the proper structures
			$prev_clause=false;
			$real_pos=0;
			$shard_query_item = "";
			$coord_query_item = "";

			$is_group_by_item = $this->process_undistributable_select_item($pos,$real_pos,$clause, $shard_query_item, $coord_query_item, $custom_functions, null, $state);

			if($shard_query) $shard_query .= ",";
			if($coord_query) $coord_query .= ",";

			$shard_query .= $shard_query_item;
			$coord_query .= $coord_query_item . " AS " . $alias;


			$strip_item_at = strpos($shard_query_item, " AS `");
			$strip_item = substr($shard_query_item, 0,$strip_item_at);

			if($is_group_by_item) {
				$group_items[] = array( 'expr' => $strip_item, 'alias' => $alias );
			}
			
		}

		$sql = "SELECT ";
		if($straight_join) $sql .= "STRAIGHT_JOIN ";
		if($distinct) $sql .= " DISTINCT ";

		$shard_query = $sql . $shard_query;
		$coord_query = $sql . $coord_query;
	
		$return = array(
				'error'         => array(),
				'shard_sql'     => $shard_query, 
				'coord_odku'    => array(),
				'coord_sql'     => $coord_query,
				'shard_group'   => "",
				'coord_group'   => "",
				'group_aliases' => "",
				'custom_functions' => $custom_functions
		);

		$gb_hash = "";

	
		if($group_items != "") {	
			foreach($group_items as $item) {
				if($gb_hash) $gb_hash .= ",";
				$gb_hash .= $item['expr'];
			}
		} else {
			#highlander reference.  The gb_hash here can be any constant value
			#because there can be only one output row
			$gb_hash = "'" . dechex(crc32("there can be only one")) . "'";
		}

		$gb_hash = "SHA1(CONCAT_WS('#'," . $gb_hash . "))";
		$return['coord_sql'] = str_replace('#gb_hash#',$gb_hash, $return['coord_sql']);
		$return['shard_sql'] = trim($return['shard_sql'],", ") . "," . $gb_hash . " as `gb_hash`";
	
		return($return);
	}
	
	protected function process_from($tables, &$state) {

		/* DEPENDENT-SUBQUERY handling
		*/

		$state->tables = array();

		foreach($tables as $key => $table) {

			if($table['expr_type']  == 'subquery') {
                if(!isset($state->tmp_shard)) $state->tmp_shard = $state->shards[array_rand($state->shards)]; 
                $sub_state = ShardQuery::new_state();
                
                #initialize the new state with a call to set_schema
                $this->set_schema($this->schema_name, $sub_state);
                $sub_state->tmp_shard = $state->tmp_shard;
			    $sub_table_name = "aggregation_tmp_" . mt_rand(1, 100000000);
			    $sub_state->table_name = $sub_table_name;
		        $sub_state->DAL = SimpleDAL::factory($sub_state->tmp_shard);

                #print_r($table['sub_tree']);

                $this->query($table['sub_tree'], false, $sub_state, true);

				$state->extra_tables[] = $sub_table_name;
                $state->extra_tables = array_merge($state->extra_tables, $sub_state->extra_tables);

				$sql = '(' . $sub_state->coord_sql . ')';
				$tables[$key]['table'] = $sql;

			} else {
				$state->tables[$table['table']] = 1;
			}
		}
		if(empty($tables[0]['alias']) || $tables[0]['alias'] == "") {
			$alias = $tables[0]['table'];
		} else {
			$alias = $tables[0]['alias']['name'];
		}
		#escape the table name if it is unescaped
		if($alias[0] != '`' && $alias[0] != '(') $alias = '`' . $alias . '`';
		if($alias != "`dual`") {
			$alias = " AS $alias";
		} else {
			$alias = " ";
		}

		#the first table is always prefixed by FROM
		$sql = "FROM " . $tables[0]['table'] . $alias;
		$cnt = count($tables);

		#now create the rest of the FROM clause
		for($i=1;$i<$cnt;++$i) {

			if($tables[$i]['ref_type'] == 'USING') {
				$tables[$i]['ref_clause']="(" . trim($tables[$i]['ref_clause']) . ")";
			} elseif($tables[$i]['ref_type'] == 'ON') {
				$tables[$i]['ref_clause'] = ' (' . $tables[$i]['ref_clause'] . ") ";
			}

			if($sql) $sql .= " ";
			$alias = $tables[$i]['alias'];
			if(empty($alias) || $alias == "") $alias = $tables[$i]['table']; else $alias=$tables[$i]['alias']['name'];

			if($alias[0] != '`' && $alias[0] != '(') {
				$pos = strpos($alias, '.');
				if($pos !== false) {
					$info = explode('.', $alias);
					$table = $info[1];
					$alias = '`' . $table . '`';	
				} else {
					$alias = '`' . $alias . '`';	

				}
			}
			if($alias != "`dual`") {
				$alias = " AS $alias";
			} else {
				$alias = " ";
			}
		    if($tables[$i]['join_type'] == 'CROSS') $tables[$i]['join_type'] = ' CROSS JOIN ';
			$sql .= $tables[$i]['join_type'] . ' ' . $tables[$i]['table'] . $alias . ' ' . $tables[$i]['ref_type'] . $tables[$i]['ref_clause'];

		}

		return $sql;
	}

	function set_shard_mapper($column, $callback) {
		$this->shard_key = $column;
		if(!is_callable($callback)) {
			throw new Exception('Invalid callback (is_callable failed)');
		}

		$this->callback = $callback;
	}

	protected function map_shard($column, $key,$schema, $operator="=", $create=false) {
		$result = call_user_func($this->callback, $column, $key, $schema, $operator, null, $create);

		if($result) return array_values($result);

        return false;
	}

	protected function append_all(&$queries, $append) {
		for($i=0;$i<count($queries);++$i) {
			$queries[$i] .= $append;
		}
	}

	protected function process_where($where, &$state = null) {
		$state->in_lists = array();
		$prev = "";
		$next_is_shard_key = false;
		$state->force_shard = array();
		$shard_id = false;
		$total_count=0;

		$sql = "WHERE";
		$queries = array($sql);

		$start_count=count($where);
        $shard_key_operator = null;
        $shard_lookup_map = array();
		foreach($where as $pos => $clause) {
			if(empty($where[$pos])) continue;
			$sql .= " ";
			$this->append_all($queries, " ");
			if($next_is_shard_key) {
				if(!trim($clause['base_expr'])) continue;
				if($clause['expr_type'] == 'const' || $clause['expr_type'] == 'in-list') {
                    $shard_lookup_map[] = array ( 'operator' => $shard_key_operator, 'value' => $clause['base_expr']);
				}
				$next_is_shard_key = false;
			}

			if ($clause['expr_type'] == 'operator')  {
				if(strtolower($clause['base_expr']) == 'between' &&
				$this->between_opt && ($this->between_opt == '*' || $this->between_opt == $prev) ) {
					$offset=0;
					$operands=array();
					#find the operands to the between expression	
					$and_count=0;

					for($n=$pos+1;$n<$start_count;++$n) {
						if($where[$n]['expr_type'] == 'operator' && strtoupper($where[$n]['base_expr']) == 'AND') {
							if($and_count) {
								break;
							} else {
								$and_count+=1;
							}
						} 
						$operands[$offset]=array('pos'=>$n,'base_expr' => $where[$n]['base_expr']);
						++$offset;
					}

					#determine what kinds of operands are in use
					$matches = $vals = array();
					$is_date = false;

					if(is_numeric(trim($operands[0]['base_expr'])) ||
					 preg_match("/('[0-9]+-[0-9]+-[0-9]+')/", $operands[0]['base_expr'], $matches)) {
						if($matches) {
							$vals[0] = $matches[0];
							$matches = array();
					 		preg_match("/('[0-9]+-[0-9]+-[0-9]+')/", $operands[2]['base_expr'], $matches);
							$vals[1] = $matches[0];
			
							$is_date =true;
						} else {
							$vals[0]  = $operands[0]['base_expr'];
							$vals[1]  = $operands[2]['base_expr'];

						}
						if(!$is_date) {
							$sub_tree = array();
							for($n=$vals[0];$n<=$vals[1];++$n) {
								$sub_tree[] = $n;
							}

						} else {

							#conversion of date between requires connecting
							#to the database to make sure that the date_diff calculation
							#is accurate for the timezone in which the database servers are
						
							$date_sql = "SELECT datediff(" . $vals[1] . ',' . $vals[0] . ") as `d`";
							$this->messages[] = "Used database to do date calculation:\n$date_sql\n\n";

							$stmt = $state->DAL->my_query($date_sql);
							if(!$stmt) {
								$this->errors[] = "Error computing date difference: " . $state->DAL->my_error();
								return false;
							}

							$row = $state->DAL->my_fetch_assoc();
							$days = $row['d'];
							for($n=0;$n<=$days;++$n) {
								$sub_tree[] = $vals[0] . " + interval $n day";
							}
						}

						for($n=$pos+1;$n<=$operands[2]['pos'];++$n) {
							unset($where[$n]);
						}

						$this->messages[] = "A BETWEEN has been converted to an IN list with " . count($sub_tree) . " items\n";
						$state->in_lists[] = $sub_tree;
						$old = $queries;

						$queries = array("");
						$ilist = "";
						$sub_tree=array_values($sub_tree);

						if(count($sub_tree) >= $this->inlist_merge_threshold ) {
							for($z=0;$z<count($sub_tree);++$z) {
								if($ilist) $ilist .= ",";
								$ilist .= $sub_tree[$z];
								if((($z+1) % $this->inlist_merge_size)  == 0 ) {
									foreach($old as $sql) {
										$queries[] = $sql . " IN (" . $ilist . ")";
									}
									$ilist= "";
								}
							}
							foreach($old as $sql) {
								if($ilist) $queries[] = $sql . " IN (" . $ilist . ")";
							}
							$ilist= "";
						} else {
							foreach($sub_tree as $val) {
								foreach($old as $sql) {
									$queries[] = $sql .= " = $val";
								}
							}	
						}

						unset($sub_tree);

						continue;
					} else {
						$this->messages[] = "BETWEEN could not be optimized - invalid operands";
					}

				} elseif($clause['expr_type'] == 'operator' /*($clause['base_expr'] == '=' || strtolower($clause['base_expr']) == 'in') */ &&
					($this->shard_key && strtolower($this->shard_key) == strtolower($prev) ))
			 	{
                    $shard_key_operator = $clause['base_expr'];
					$next_is_shard_key=true;
				}
				$this->append_all($queries, $clause['base_expr']);
			} elseif($clause['expr_type'] != 'in-list') {
				$this->append_all($queries, $clause['base_expr']);
				$prev = $clause['base_expr'];

			} elseif($this->inlist_opt && ($this->inlist_opt == '*' || $this->inlist_opt == $prev)) {

				$old = $queries;
				$queries=array();
				
				foreach($clause['sub_tree'] as $vals) {
					foreach($old as $sql) {
						$queries[] = "$sql (${vals['base_expr']}) ";
					}
				}	

			} else {
				$prev = $clause['base_expr'];
				$this->append_all($queries, $prev);
			}
			
		}

        $output_map = array();
        if(!empty($shard_lookup_map)) {
            foreach($shard_lookup_map as $lookup) {
                $shard_ids = $this->map_shard($this->shard_key, $lookup['value'], $this->current_schema, $lookup['operator']);
                if($shard_ids == null) {
                    #map failed (invalid operator?) must broadcast!
                    $output_map = null;
                    break;
                }

                if($shard_ids == false) {
                    #mapping simply was not found
                    continue;
                }

                foreach($shard_ids as $shard) {
                   $shard_name = array_pop(array_keys($shard));
                   $output_map[$shard_name] = $this->shards[$shard_name];
                   $this->messages[] = "Mapped query to shard: $shard_name";
                }
            }
        }


        if(empty($output_map)) $state->force_shard = null;
        $state->force_shard = $output_map;

		foreach($queries as $pos => $q) {
			if(!trim($q)) unset($queries[$pos]);
		}

		return array_values($queries);
	}

	/* if $sql is an Array(), then it is assumed it is already parsed */	
	protected function process_sql($sql, &$state) {

		#only useful for the fetch worker for debugging	
		$state->sql = $sql;
		$parser=null;
		$straight_join = false;
	
		$state->shard_sql = ""; #identical SQL which will be broadcast to all shards
		$state->coord_sql = ""; #the summary table sql
		$state->in_lists=array()	;

		$state->used_agg_func = false;

		$error = array();
	
		$select = null;
				
		if(!is_array($sql)) {
			$blacklist = 
			'/(?:create|drop|alter)\s+database|last_insert_id|sql_calc_found_rows|row_count|:=|connection_id|' .
			'current_user|session_user|user\(\)|mysql\.|system_user|found_rows|get_lock|free_lock|' .
			'is_free_lock|is_used_lock|load_file|infile|master_pos_wait|name_const|release_lock|^show\s+(?:master|slave)|' .
			'^show\s+bin|^show\s+create\s+database|^show\s+databases|^show\s+logs|^show\s+mutex|' .
			'^show\s+engine|^show\s+innodb|^show\s+open|^show\s+privileges|^show\s+(?:status|variables)|^lock\s+table|' .
			'^check\s+table|^checksum\s+table|^backup|^repair|^restore|^call|^handler|^begin|^start\s+transaction|^commit|' .
			'^rollback|^set transaction|savepoint|^xa\s+(?:start|begin|end|prepare|commit|rollback|recover)/i';

			if(preg_match($blacklist, $sql,$matches)) {
				throw new Exception('Unsupported SQL: ' . print_r($matches, true));
			}

			#we need to check the syntax
			$sql = preg_replace(array("%-- [^\\n](?:\\n|$)%","%/\\*.*\\*/%"), '', $sql);
			if(preg_match('/^select/i', $sql)) {
				$check_sql = $sql;
				#custom functions have to pass the syntax check.  concat takes a variable number of 
				#parameters, so it can be substituted for any custom function for syntax purposes
				foreach(ShardQueryCustomFunctions::$list as $key => $val) {
					$check_sql = preg_replace('/' . $key . '/i', 'concat', $check_sql); 
				}
				$check_sql = "select count(*) from (" . trim($check_sql,";\r\n ") . ") check_sql where 0=1";
				if(!$state->DAL->my_query($check_sql)) throw new Exception($state->DAL->my_error());
			}

			$state->parsed = $this->parser->parse($sql);
		} else {
			$state->parsed = $sql;
		}


		if(!empty($state->parsed['UNION ALL'])) {
			$queries = array();
			$table_name = "aggregation_tmp_" . mt_rand(1, 100000000);
            $state->table_name = $table_name;

            $coord_sql = "";
			foreach($state->parsed['UNION ALL'] as $sub_tree) {
                $sub_state = ShardQuery::new_state();
                
                #initialize the new state with a call to set_schema
                $this->set_schema($this->schema_name, $sub_state);
                $sub_state->tmp_shard = $state->tmp_shard;
			    $sub_table_name = "aggregation_tmp_" . mt_rand(1, 100000000);
			    $sub_state->table_name = $sub_table_name;
		        $sub_state->DAL = SimpleDAL::factory($sub_state->tmp_shard);
                $this->query($sub_tree, false, $sub_state, true);
				$state->extra_tables[] = $sub_table_name;
                $state->extra_tables = array_merge($state->extra_tables, $sub_state->extra_tables);

                if($coord_sql) $coord_sql .= " UNION ALL ";
                $coord_sql .= "( " . $sub_state->coord_sql . " )";
				$state->extra_tables[] = $sub_table_name;
    
			}	

			$select['coord_sql'] = $coord_sql;
            unset($coord_sql);
		} elseif(!empty($state->parsed['UNION'])) {
			$queries = array();
			$table_name = "aggregation_tmp_" . mt_rand(1, 100000000);
            $state->table_name = $table_name;

            $coord_sql = "";
			foreach($state->parsed['UNION'] as $sub_tree) {
                $sub_state = ShardQuery::new_state();
                
                #initialize the new state with a call to set_schema
                $this->set_schema($this->schema_name, $sub_state);
                $sub_state->tmp_shard = $state->tmp_shard;
			    $sub_table_name = "aggregation_tmp_" . mt_rand(1, 100000000);
			    $sub_state->table_name = $sub_table_name;
		        $sub_state->DAL = SimpleDAL::factory($sub_state->tmp_shard);
                $this->query($sub_tree, false, $sub_state, true);
				$state->extra_tables[] = $sub_table_name;
                $state->extra_tables = array_merge($state->extra_tables, $sub_state->extra_tables);

                if($coord_sql) $coord_sql .= " UNION ";
                $coord_sql .= "( " . $sub_state->coord_sql . " )";
    
			}	
			#UNION operation requires deduplication of the temporary table
			$select['coord_sql'] = $coord_sql;
            unset($coord_sql);
		} elseif(!empty($state->parsed['SELECT']) && empty($state->parsed['INSERT']) && empty($state->parsed['CREATE'])) {
			#reset the important variables	
			$select = $from = $where = $group = $order_by = "";
			$straight_join = $distinct = false;
			$this->errors=array();
			
			#The SELECT clause is processed first.
			if(!empty($state->parsed['OPTIONS']) && in_array('STRAIGHT_JOIN', $state->parsed['OPTIONS'])) { 
				$straight_join=true;
			}

			if(!empty($state->parsed['OPTIONS']) && in_array('DISTINCT', $state->parsed['OPTIONS'])) {
				$distinct = true;
			}

			if(!empty($state->parsed['OPTIONS']) && in_array('SQL_CALC_FOUND_ROWS', $state->parsed['OPTIONS'])) {
				$this->errors[] = array('Unsupported query', 'SQL_CALC_FOUND_ROWS is not supported');
				return false;
			}

			if(!empty($state->parsed['OPTIONS']) && in_array('FOR UPDATE', $state->parsed['OPTIONS'])) {
				$this->errors[] = array('Unsupported query', 'FOR UPDATE is not supported');
				return false;
			}

			if(!empty($state->parsed['OPTIONS']) && in_array('LOCK IN SHARE MODE', $state->parsed['OPTIONS'])) {
				$this->errors[] = array('Unsupported query', 'LOCK IN SHARE MODE is not supported');
				return false;
			}

			$with_rollup = "";
			if(!empty($state->parsed['OPTIONS']) && in_array('WITH ROLLUP', $state->parsed['OPTIONS'])) {
				$with_rollup = ' WITH ROLLUP';
			}

			//ignore any other options
			unset($state->parsed['OPTIONS']);

			$procedure = "";
			if(!empty($state->parsed['PROCEDURE'])) {
				$procedure = 'PROCEDURE ' . join('', $state->parsed['PROCEDURE']);
				unset($state->parsed['PROCEDURE']);
			}

			if(!empty($state->parsed['INTO'])) {
				$this->errors[] = array('Unsupported query', 'INTO is not supported');
				return false;

			}

			/*This will process the SELECT clause, doing any SQL rewrites needed for aggregation at the shard level
			  The $this->non_distrib flag will be set if the query contains any non-distributable aggregate functions
			  such as STDDEV(some_col).

			  If non-distributable aggregate functions are detected, we will throw this work away and instead use
			  $this->process_undistributable_select() [below]
			*/
			$select = $this->process_select($state->parsed['SELECT'], $straight_join,$distinct, $state);

			if(!empty($select['error'])) {
				$this->errors[] = $select['error'];
				return false;
			}

			if(!$state->non_distrib) { 
				unset($state->parsed['SELECT']);
			} else {
				$select = $this->process_undistributable_select($state->parsed['SELECT'], $straight_join, $distinct);
				unset($state->parsed['SELECT']);
			}

			if(!empty($state->parsed['GROUP']) && empty($select['coord_group'])) {
				foreach($state->parsed['GROUP'] as $gb) {
					if(!is_numeric(trim($gb['base_expr']))) { 
						$id = "expr_" . mt_rand(1, 100000);
					
						$select['shard_sql'] .= "," . $gb['base_expr'] . ' AS `' . $id . '`';
						if($select['shard_group'] !== "") $select['shard_group'] .= ',';
						$select['shard_group'] .= $gb['base_expr'];

						if($select['coord_group'] !== "") $select['coord_group'] .= ',';
						$select['coord_group'] .= "`" . $id . "`";
					}
				}
			}


			unset($state->parsed['GROUP']);

			if($having_info = $this->process_having($state)) {
				if(trim($having_info['select'])) $select['shard_sql'] .= "," . $having_info['select'];
			} else {
				$having_info = array('having' => " ");
			}		


			
			if(empty($state->parsed['FROM'])) {
				$this->errors = array('Unsupported query', 'Missing FROM clause');
				return false;
			} else {
				$select['shard_sql'] .= "\n" . $this->process_from($state->parsed['FROM'], $state);
			
				$tables = array_keys($state->tables);
				$in_tables = "";
				foreach($tables as $val) {
					if($in_tables !== "") $in_tables .= ",";
					$in_tables .= "'{$val}'";
				}

                if($in_tables !== "") {
				    $sql = "select count(*) cnt from information_schema.columns where table_schema='" . 
				    	$state->tmp_shard['db'] . "' and column_name='" . $this->shard_key .
				    	"' and table_name IN({$in_tables});";

				    if(!$stmt = $state->DAL->my_query($sql)) throw new Exception('could not access information schema: ' . $state->DAL->my_error());
				    $row=$state->DAL->my_fetch_assoc();

                    /* If none of the tables contain the sharding key, the query goes to all shards */
				    if($row['cnt'] === '0') {
				    	$state->force_shard = array($state->shards[array_rand($state->shards)]);
				    } 
                } else {
                    $state->force_shard = array($state->tmp_shard);
                } 

				if(!isset($state->table_name) || $state->table_name == "") $state->table_name = "aggregation_tmp_" . mt_rand(1, 100000000);
				$select['coord_sql'] .= "\nFROM `$state->table_name`";
	
				unset($state->parsed['FROM']);
			}

			#note that this will extract inlists and store them in $this->in_lists (if inlist optimization is on) 	
			if(!empty($state->parsed['WHERE'])) {
				$where_clauses = $this->process_where($state->parsed['WHERE'], $state);
				unset($state->parsed['WHERE']);
			}

			if(!empty($state->parsed['ORDER'])) {
				$order_by = "";
				foreach($state->parsed['ORDER'] as $o) {
					if($order_by) $order_by .= ',';

					$order_by .= $o['base_expr'];

					$item = "";
					switch($o['expr_type']) {
						case 'function':
						case 'aggregate_function':
						case 'expression':
							$this->concat_all_subtrees($o['sub_tree'],$item);
							$order_by .= "($item)";
						break;
					}
					$order_by .= " " . $o['direction'];
				}	
				$order_by = " ORDER BY {$order_by}";
				unset($state->parsed['ORDER']);
			}

			$used_limit = 0;
			if(!empty($state->parsed['LIMIT']) ) {
				$used_limit = 1;
				if($state->parsed['LIMIT']['offset'] == "") $state->parsed['LIMIT']['offset'] = 0;
				$order_by .= " LIMIT {$state->parsed['LIMIT']['offset']},{$state->parsed['LIMIT']['rowcount']}";
				unset($state->parsed['LIMIT']);
			}

			

	
			foreach($state->parsed as $key => $clauses) {
				$this->errors[] = array('Unsupported query',$key . ' clause is not supported');
			}

	
			if($this->errors) {
				return false;
			}

			
			#process_select only provides a list of positions
			if($select['coord_group']) $select['coord_group'] = ' GROUP BY ' . $select['coord_group'] . $having_info['having'] . $with_rollup . $procedure;
			if($select['shard_group']) $select['shard_group'] = ' GROUP BY ' . $select['shard_group'];

			$queries = array();
			if(!empty($where_clauses)) {
				foreach($where_clauses as $where) {
					if(!$state->no_pushdown_limit) {
						$queries[] = $select['shard_sql'] . ' ' . $where . ' ' . $select['shard_group'] . ($used_limit ? $order_by : ($state->used_agg_func ? ' ORDER BY NULL' : ''));
					} else {
						$queries[] = $select['shard_sql'] . ' ' . $where . ' ' . $select['shard_group'] . ($state->used_agg_func ? ' ORDER BY NULL' : '');

					}
				}
			} else {
				if(!$state->no_pushdown_limit) {
					$queries[] = $select['shard_sql'] . $select['shard_group'] . ($used_limit ? $order_by : ($state->used_agg_func ? ' ORDER BY NULL' : ''));
				} else {
					$queries[] = $select['shard_sql'] . $select['shard_group'] . ($state->used_agg_func ? ' ORDER BY NULL' : '');
				}
			}



		} elseif(!empty($state->parsed['INSERT'])) { 

			if(!empty($state->parsed['SELECT'])) {
				$state->broadcast_query = $state->orig_sql;
			} else {
				$table_name = $state->parsed['INSERT']['table'];

				if($state->parsed['INSERT']['columns'] == '') {
					if(empty($this->col_metadata_cache[$table_name])) {
						$sql = "select column_name 
							  from information_schema.columns 
						  	 where table_name='{$table_name}'
							   and table_schema='{$this->tmp_shard['db']}' 
						    	 order by ordinal_position;";

						if(!$stmt=$state->DAL->my_query($sql)) throw new Exception($state->DAL->my_error());
						$cols = array();
						$col_list = "";
						while($row = $state->DAL->my_fetch_assoc($stmt)) {
							if($col_list) $col_list .= ",";
							$col_list .= $row['column_name'];
							$cols[] = $row['column_name'];
						}
						if(count($cols) == 0) {
							throw new Exception('Table not found in data dictionary');
						}
						$this->col_metadata_cache[$table_name] = array('cols' => $cols, 'col_list' => $col_list);
					} else { 
						$cols = $this->col_metadata_cache[$table_name]['cols'];
						$col_list = $this->col_metadata_cache[$table_name]['col_list'];
					}

				} else {
					#FIXME: this can't be right
					$cols = $state->parsed['INSERT']['columns'];
					print_r($cols);
					exit;
				}

				$shard_key_pos = false;	
				if(empty($this->table_is_sharded_cache[$table_name])) {
					foreach($cols as $pos => $col) {
						if(trim($col,'` ') == $this->shard_key) {
							$shard_key_pos = $pos;
							break;
						}	
					}

					$this->table_is_sharded_cache[$table_name] = $shard_key_pos;
				} else { 
					$shard_key_pos = $this->table_is_sharded_cache[$table_name];
				}
		 
				$sql = "INSERT INTO `" . $state->parsed['INSERT']['table'] . "` ({$col_list}) VALUES ";
				$values = array();
				$val_count = 0;

				$column_count = count($cols);
				
				foreach($state->parsed['VALUES'] as $record) {

					$vals = $record['base_expr'];
		
					#this row is going to go to a single shard
					if($shard_key_pos !== false) {
						if(count($record['data']) !== $column_count) {
							throw new Exception('Value count does not match column count!');
						}	
						$shard_id=$this->map_shard($this->shard_key, $record['data'][$shard_key_pos]['base_expr']);

						if(empty($values[$shard_id])) {
							$values[$shard_id] = $vals;	
						} else {
							$values[$shard_id] .= ",{$vals}";
						}

						$val_count++;

					#the row is going to all shards
					} else {
						foreach($this->shards as $shard_id => $shard) {
							if(empty($values[$shard_id])) {
								$values[$shard_id] = $vals;	
							} else {
								$values[$shard_id] .= ",{$vals}";
							}

							$val_count++;
						}

					}
				}

				$state->insert_sql = $sql;
				$state->insert_values = $values;

			}

			return true;

		} else {
			//This query should be broadcast to all nodes
			$state->broadcast_query = $state->orig_sql;
			return true;
		}

	
        #if(!isset($state->table_name)) $state->table_name = "agg_	
		if(empty($order_by)) $order_by = "";
		if(empty($select['coord_group'])) $select['coord_group'] = "";
		if(empty($select['coord_odku'])) $select['coord_odku'] = "";
		if(empty($select['group_aliases'])) $select['group_aliases'] = "";
		$state->coord_sql = $select['coord_sql'] . ' ' . $select['coord_group'] . ' ' . $order_by;
		if(!isset($state->coord_odku)) $state->coord_odku = $select['coord_odku'];
		$state->shard_sql = $queries;

		$state->agg_key_cols = $select['group_aliases'];

		if(!empty($select['custom_functions'])) {
			$state->custom_functions = $select['custom_functions'];
		} else {
			$state->custom_functions = array();
		}
	
		$explain = "Shard-Query optimizer messages:";
		if($state->agg_key_cols) { 
			$explain .= "\n  * The following projections may be selected for a UNIQUE CHECK on the storage node operation:\n  {$state->agg_key_cols}\n";
		        if($state->coord_odku) $explain .= "\n  * storage node result set merge optimization enabled:\n  ON DUPLICATE KEY UPDATE\n\t" . join(",\n\t",$state->coord_odku);
		}

		if(isset($state->messages)) {
            foreach($state->messages as $msg) {
			    $explain .= "\n  * $msg";
		    }
        }

		if($state->non_distrib) {
			$explain .= "\n  * Aggregation pushdown disabled due to use of non-distributable and/or custom functions.";
		}

		$explain .= "\n\nSQL TO SEND TO SHARDS:\n";
		$explain .= print_r($state->shard_sql,true);

		$explain .= "SQL TO SEND TO COORDINATOR NODE:\n{$state->coord_sql}\n";

		$state->explain = $explain;

		return true;
	
	}

	protected function create_gearman_set($queries, $table, $complete_cb=null,$failure_cb=null, &$set, $action = "store_resultset", &$state) {
        if($this->async) {
            $task_type = Net_Gearman_task::JOB_BACKGROUND;
            $query_id = $state->query_id;
        } else {
            $task_type = Net_Gearman_task::JOB_NORMAL;
            $query_id = null;
        }

		$sets = array();
		if(!$set) $set = new Net_Gearman_Set();
        if(isset($state->force_shard) && !empty($state->force_shard)) $shards = $state->force_shard; else $shards = $state->shards;
    
		for($i=0; $i<count($queries); ++$i) {
	  		foreach($shards as $shard ) {
	  			$task = new Net_Gearman_Task(   "store_resultset",
								array(  'engine'=>$state->engine,
									'table_name' => $table, 
									'sql' => $queries[$i],
									'coord_odku'=>$state->coord_odku, 
									'shard'=>$shard,
									'tmp_shard'=>$state->tmp_shard, 
									'agg_key_cols'=> $state->agg_key_cols, 
									'when' => microtime(true),
									'action' => $action,/*
									'custom_functions' => $custom_functions*/
                                    'query_id' => $query_id 
								),
								uniqid(md5($queries[$i]),true),
								$task_type
								
				);
				$task->attachCallback($complete_cb);
				$task->attachCallback($failure_cb,Net_Gearman_Task::TASK_FAIL);
				$set->addTask($task);
                $state->store_resultset_count++;
			}
			$sets[] = $set;
		}	

		return $sets;
	}

	protected function run_set($set, &$state) { 
		if(!is_array($set)) $set = array($set);

		foreach($set as $the_set) {	
			$this->workers++;
			$state->client->runSet($the_set);
		}
	}


	protected function aggregate_result(&$state = null) {
		if(!$state->DAL->my_select_db($state->tmp_shard['db'])) throw new Exception($state->DAL->my_error());

		if(!$stmt = $state->DAL->my_query($state->coord_sql)) { 
			$this->errors[] = $state->DAL->my_error();
			return false;
		}

		$this->cleanup_tables(&$state);

		return $stmt;
	
	}

	function cleanup_tables(&$state = null) {
        if(!isset($state)) $state = $this->state;

		$sql = "DROP TABLE IF EXISTS {$state->table_name};";
		@$state->DAL->my_query($sql);

		foreach($state->extra_tables as $table) {
			$sql = "DROP TABLE IF EXISTS {$table};";
			@$state->DAL->my_query($sql);
		}
	}

	#callback that records that one of the workers failed	
	function handle_failure($func) {
		--$this->workers;
		if(!$this->errors) $this->errors=array();
		$this->errors[] = "WORKER_FAIL: handle_failure registered unsuccessful worker completion.\n";
	}

	function generic_callback($func, $handle, $result) {
		--$this->workers;
		if(!$this->errors) $this->errors=array();
		if(!$this->info) $this->info=array();
		if(is_array($result) && !empty($result['error'])) {
			$this->errors[] = $result['error'];
		}
		if(is_array($result) && !empty($result['info'])) {
			$this->info[] = $result['info'];
		}
		if(!empty($result['done']) && $result['done'] > 0) $this->state->got_rows += $result['done'];
		if(!empty($result['hashes']['gb_hash'])) {
			#dedupe the hashes using an associative array
			foreach(array_keys($result['hashes']['gb_hash']) as $hash) {
				if(empty($this->hashes[$hash])) {
					$this->hashes[$hash] = 1;
				} else {
					$this->hashes[$hash]++;
				}
			}
		}
	}
	
}
